package com.chiry.business.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.Optional;

@Component
public class AcceptKafkaStream {
    private static final Logger LOG = LoggerFactory.getLogger(AcceptKafkaStream.class);
    public static final String topic = "topic-snap-image";

    //@KafkaListener(topics = topic)
    public void consumer(ConsumerRecord consumerRecord) {
        Optional<Object> kafkaMassage = Optional.ofNullable(consumerRecord.value());
        if (kafkaMassage.isPresent()) {
            Object o = kafkaMassage.get();
            LOG.info(o.toString());
        }
        LOG.info("offset: {} 收到消息: key === {} ", consumerRecord.offset(), new String(consumerRecord.value().toString().getBytes(), StandardCharsets.UTF_8));

    }
}
